#include<stdlib.h>
#include<thread>
#include<mutex>
#include<iostream>
#include<condition_variable>
#include<queue>
#include<memory>
#include<vector>

using namespace std;

class Task
{

public:
    Task()
    {
    }
    ~Task()
    {}
    virtual void process()=0;

private:

};

class threadPool;


class taskQueue
{
public:
    taskQueue(int n)
        :_queSize(n)
         ,_quit(false)
    {

    }

    friend class threadPool;

    //添加任务

    void push(Task* task)
    {
        unique_lock<mutex> lock(_mutex);
        while(full())
        {
            _notFull.wait(lock);
        }
        _que.push(task);
        _notEmpty.notify_one();
    }

    Task* pop()
    {
        unique_lock<mutex> lock(_mutex);
        while(empty()&&!_quit)
        {
            _notEmpty.wait(lock);
        }
        if(!_quit)
        {
            auto pt = _que.front();
            _que.pop();
            _notFull.notify_one();
            return pt;
        }
        else
        {
            return nullptr;
        }
    }


    //任务队列判断空
    bool empty()
    {
        return _que.size()==0;
    }

    //任务队列判断满
    bool full()
    {
        return _que.size()==_queSize;
    }

    void wakeup()
    {
        _quit=true;
        _notEmpty.notify_all();
    }

private:
    size_t _queSize;
    queue<Task*> _que;
    condition_variable _notFull;
    condition_variable _notEmpty;
    mutex _mutex;
    bool _quit;
};

class threadPool
{
public:
    threadPool(size_t n,size_t quesize)
        :_threadNum(n)
         ,_queSize(quesize)
         ,_que(quesize)
         ,_isexit(false)
    {

    }

    ~threadPool()
    {}

    friend class MyTask;


    void start()
    {
        //线程池参数传递之后，启动现成
        for(int  i=0;i<_threadNum;++i)
        {
            _threads.push_back(thread(&threadPool::doTask,this));
        }
    }

    void stop()
    {

        while(!_que.empty())
        {

            std::this_thread::sleep_for(std::chrono::seconds(1));

        }

        _isexit=true;

        _que.wakeup();

        for(auto& th:_threads)
        {

            th.join();

        }
    }

    void addTask(Task* lhs)
    {
        if(!_isexit)
        {
            _que.push(lhs);
        }
    }

    Task* getTask()
    {
        auto func= _que.pop();
        return func;
    }

    void doTask()
    {
        while(!_isexit)
        {
            Task* func = getTask();
            if(func)
            {
                func->process();
            }
            else
            {
                cout<<"null_ptr\n";
            }
        }
    }

private:
    size_t _threadNum;
    size_t _queSize;
    vector<thread> _threads;
    taskQueue _que;
    bool _isexit;
    static mutex _outputmutex;
};

mutex threadPool::_outputmutex;
class MyTask
:public Task
{
public:
    MyTask()
    {

    }
    ~MyTask()
    {
    }
    void  process() override
    {
        ::srand(::clock());
        int number = ::rand()%100;
        unique_lock<mutex> lock(threadPool::_outputmutex);
        cout<<this_thread::get_id()<<"  MyTask process =  "<<number<<"\n";
    }
private:
};

int main()
{
    threadPool pool(4,4);
    unique_ptr<Task> ptask(new MyTask());
    pool.start();
    for(int i=0;i<100;++i)
    {
        pool.addTask(ptask.get());
    }
    pool.stop();
    return 0;
}
